package com.fs.producer;

import com.fs.entity.Demo06Message;
import com.fs.entity.JudgeItem;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

/**
 * @author ：hyq
 * @date ：Created in 2020/5/20 14:15
 * @description：
 * @modified By：
 */
@Component
public class Producer {
    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;
    private Gson gson = new GsonBuilder().create();


    public void send(Integer i){
        JudgeItem judgeItem = JudgeItem.builder()
                .metric("下载文件")
                .type(2)
                .content("文件id为n2658dcd8载失败")
                .endPoint("文件服务")
                .build();
         kafkaTemplate.send("alarm",i.toString(), gson.toJson(judgeItem));
    }


    /**
     * 测试只定分区顺序消费
     * @param id
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(Integer id) throws ExecutionException, InterruptedException {
        // 创建 Demo01Message 消息
        Demo06Message message = new Demo06Message();
        message.setId(id);
        // 同步发送消息
        // 因为我们使用 String 的方式序列化 key ，所以需要将 id 转换成 String
        return kafkaTemplate.send(Demo06Message.TOPIC, String.valueOf(id), message).get();
    }

    /**
     * 测试kafka 结合 spring 注解事务发送
     * @throws InterruptedException
     */
    @Transactional
    public void testTransactionalAnnotation() throws InterruptedException {
        kafkaTemplate.send("alarm", "test transactional annotation");
        int i = 1/0;
    }




}
